//===----------------------------------------------------------------------===//
//                         DuckDB
// This file will not be overwritten. To implement a custom function for
// a new setting, enable 'custom_implementation' in 'src/common/settings.json'
// for this setting. The 'update_settings_definitions.py' may include new
// setting methods' signatures that need to be implemented in this file. You
// can check the functions declaration in 'settings.hpp' and what is
// autogenerated in 'autogenerated_settings.cpp'.
//
//===----------------------------------------------------------------------===//

#include "duckdb/main/settings.hpp"

#include "duckdb/common/enums/access_mode.hpp"
#include "duckdb/catalog/catalog_search_path.hpp"
#include "duckdb/common/string_util.hpp"
#include "duckdb/common/operator/double_cast_operator.hpp"
#include "duckdb/main/attached_database.hpp"
#include "duckdb/main/client_context.hpp"
#include "duckdb/main/client_data.hpp"
#include "duckdb/main/config.hpp"
#include "duckdb/main/database.hpp"
#include "duckdb/main/database_manager.hpp"
#include "duckdb/main/query_profiler.hpp"
#include "duckdb/main/secret/secret_manager.hpp"
#include "duckdb/parallel/task_scheduler.hpp"
#include "duckdb/parser/parser.hpp"
#include "duckdb/planner/expression_binder.hpp"
#include "duckdb/storage/external_file_cache.hpp"
#include "duckdb/storage/buffer/buffer_pool.hpp"
#include "duckdb/storage/buffer_manager.hpp"
#include "duckdb/storage/storage_manager.hpp"
#include "duckdb/logging/logger.hpp"
#include "duckdb/logging/log_manager.hpp"
#include "duckdb/common/type_visitor.hpp"
#include "duckdb/function/variant/variant_shredding.hpp"
#include "duckdb/storage/block_allocator.hpp"

namespace duckdb {

constexpr const char *LoggingMode::Name;
constexpr const char *LoggingLevel::Name;
constexpr const char *EnableLogging::Name;
constexpr const char *LoggingStorage::Name;
constexpr const char *EnabledLogTypes::Name;
constexpr const char *DisabledLogTypes::Name;
constexpr const char *DisabledFilesystemsSetting::Name;

const string GetDefaultUserAgent() {
	return StringUtil::Format("duckdb/%s(%s)", DuckDB::LibraryVersion(), DuckDB::Platform());
}

namespace {

template <class T>
static DatabaseInstance &GetDB(DatabaseInstance *db) {
	if (!db) {
		throw InvalidInputException("Cannot change/set %s before the database is started", T::Name);
	}
	return *db;
}

} // namespace

//===----------------------------------------------------------------------===//
// Access Mode
//===----------------------------------------------------------------------===//
bool AccessModeSetting::OnGlobalSet(DatabaseInstance *db, DBConfig &config, const Value &input) {
	if (db) {
		throw InvalidInputException("Cannot change access_mode setting while database is running - it must be set when "
		                            "opening or attaching the database");
	}
	return true;
}

//===----------------------------------------------------------------------===//
// Allocator Background Threads
//===----------------------------------------------------------------------===//
bool AllocatorBackgroundThreadsSetting::OnGlobalSet(DatabaseInstance *db, DBConfig &config, const Value &input) {
	if (db) {
		TaskScheduler::GetScheduler(*db).SetAllocatorBackgroundThreads(input.GetValue<bool>());
	}
	return true;
}

bool AllocatorBackgroundThreadsSetting::OnGlobalReset(DatabaseInstance *db, DBConfig &config) {
	if (db) {
		TaskScheduler::GetScheduler(*db).SetAllocatorBackgroundThreads(DBConfigOptions().allocator_background_threads);
	}
	return true;
}

//===----------------------------------------------------------------------===//
// Allocator Bulk Deallocation Flush Threshold
//===----------------------------------------------------------------------===//
void AllocatorBulkDeallocationFlushThresholdSetting::SetGlobal(DatabaseInstance *db, DBConfig &config,
                                                               const Value &input) {
	config.options.allocator_bulk_deallocation_flush_threshold = DBConfig::ParseMemoryLimit(input.ToString());
	if (db) {
		BufferManager::GetBufferManager(*db).GetBufferPool().SetAllocatorBulkDeallocationFlushThreshold(
		    config.options.allocator_bulk_deallocation_flush_threshold);
	}
}

void AllocatorBulkDeallocationFlushThresholdSetting::ResetGlobal(DatabaseInstance *db, DBConfig &config) {
	config.options.allocator_bulk_deallocation_flush_threshold =
	    DBConfigOptions().allocator_bulk_deallocation_flush_threshold;
	if (db) {
		BufferManager::GetBufferManager(*db).GetBufferPool().SetAllocatorBulkDeallocationFlushThreshold(
		    config.options.allocator_bulk_deallocation_flush_threshold);
	}
}

Value AllocatorBulkDeallocationFlushThresholdSetting::GetSetting(const ClientContext &context) {
	auto &config = DBConfig::GetConfig(context);
	return Value(StringUtil::BytesToHumanReadableString(config.options.allocator_bulk_deallocation_flush_threshold));
}

//===----------------------------------------------------------------------===//
// Allocator Flush Threshold
//===----------------------------------------------------------------------===//
void AllocatorFlushThresholdSetting::SetGlobal(DatabaseInstance *db, DBConfig &config, const Value &input) {
	config.options.allocator_flush_threshold = DBConfig::ParseMemoryLimit(input.ToString());
	if (db) {
		TaskScheduler::GetScheduler(*db).SetAllocatorFlushTreshold(config.options.allocator_flush_threshold);
	}
}

void AllocatorFlushThresholdSetting::ResetGlobal(DatabaseInstance *db, DBConfig &config) {
	config.options.allocator_flush_threshold = DBConfigOptions().allocator_flush_threshold;
	if (db) {
		TaskScheduler::GetScheduler(*db).SetAllocatorFlushTreshold(config.options.allocator_flush_threshold);
	}
}

Value AllocatorFlushThresholdSetting::GetSetting(const ClientContext &context) {
	auto &config = DBConfig::GetConfig(context);
	return Value(StringUtil::BytesToHumanReadableString(config.options.allocator_flush_threshold));
}

//===----------------------------------------------------------------------===//
// Allow Community Extensions
//===----------------------------------------------------------------------===//
bool AllowCommunityExtensionsSetting::OnGlobalSet(DatabaseInstance *db, DBConfig &config, const Value &input) {
	if (db && !config.options.allow_community_extensions) {
		auto new_value = input.GetValue<bool>();
		if (new_value) {
			throw InvalidInputException("Cannot upgrade allow_community_extensions setting while database is running");
		}
		return false;
	}
	return true;
}

bool AllowCommunityExtensionsSetting::OnGlobalReset(DatabaseInstance *db, DBConfig &config) {
	if (db && !config.options.allow_community_extensions) {
		if (DBConfigOptions().allow_community_extensions) {
			throw InvalidInputException("Cannot upgrade allow_community_extensions setting while database is running");
		}
		return false;
	}
	return true;
}

//===----------------------------------------------------------------------===//
// Allow Parser Override
//===----------------------------------------------------------------------===//
bool AllowParserOverrideExtensionSetting::OnGlobalSet(DatabaseInstance *db, DBConfig &config, const Value &input) {
	auto new_value = input.GetValue<string>();
	vector<string> supported_options = {"default", "fallback", "strict", "strict_when_supported"};
	string supported_option_string;
	for (const auto &option : supported_options) {
		if (StringUtil::CIEquals(new_value, option)) {
			return true;
		}
	}
	throw InvalidInputException("Unrecognized value for parser override setting. Valid options are: %s",
	                            StringUtil::Join(supported_options, ", "));
}

bool AllowParserOverrideExtensionSetting::OnGlobalReset(DatabaseInstance *db, DBConfig &config) {
	config.options.allow_parser_override_extension = "default";
	return true;
}

//===----------------------------------------------------------------------===//
// Allow Persistent Secrets
//===----------------------------------------------------------------------===//
void AllowPersistentSecretsSetting::SetGlobal(DatabaseInstance *db, DBConfig &config, const Value &input) {
	auto value = input.DefaultCastAs(LogicalType::BOOLEAN);
	config.secret_manager->SetEnablePersistentSecrets(value.GetValue<bool>());
}

void AllowPersistentSecretsSetting::ResetGlobal(DatabaseInstance *db, DBConfig &config) {
	config.secret_manager->ResetEnablePersistentSecrets();
}

Value AllowPersistentSecretsSetting::GetSetting(const ClientContext &context) {
	auto &config = DBConfig::GetConfig(context);
	return Value::BOOLEAN(config.secret_manager->PersistentSecretsEnabled());
}

//===----------------------------------------------------------------------===//
// Allow Unredacted Secrets
//===----------------------------------------------------------------------===//
bool AllowUnredactedSecretsSetting::OnGlobalSet(DatabaseInstance *db, DBConfig &config, const Value &input) {
	if (db && input.GetValue<bool>()) {
		throw InvalidInputException("Cannot change allow_unredacted_secrets setting while database is running");
	}
	return true;
}

bool AllowUnredactedSecretsSetting::OnGlobalReset(DatabaseInstance *db, DBConfig &config) {
	if (db) {
		throw InvalidInputException("Cannot change allow_unredacted_secrets setting while database is running");
	}
	return true;
}

//===----------------------------------------------------------------------===//
// Disable Database Invalidation
//===----------------------------------------------------------------------===//
bool DisableDatabaseInvalidationSetting::OnGlobalSet(DatabaseInstance *db, DBConfig &config, const Value &input) {
	if (db && input.GetValue<bool>()) {
		throw InvalidInputException("Cannot change disable_database_invalidation setting while database is running");
	}
	return true;
}

bool DisableDatabaseInvalidationSetting::OnGlobalReset(DatabaseInstance *db, DBConfig &config) {
	if (db) {
		throw InvalidInputException("Cannot change disable_database_invalidation setting while database is running");
	}
	return true;
}

//===----------------------------------------------------------------------===//
// Allow Unsigned Extensions
//===----------------------------------------------------------------------===//
bool AllowUnsignedExtensionsSetting::OnGlobalSet(DatabaseInstance *db, DBConfig &config, const Value &input) {
	if (db && input.GetValue<bool>()) {
		throw InvalidInputException("Cannot change allow_unsigned_extensions setting while database is running");
	}
	return true;
}

bool AllowUnsignedExtensionsSetting::OnGlobalReset(DatabaseInstance *db, DBConfig &config) {
	if (db) {
		throw InvalidInputException("Cannot change allow_unsigned_extensions setting while database is running");
	}
	return true;
}

//===----------------------------------------------------------------------===//
// Allowed Directories
//===----------------------------------------------------------------------===//
void AllowedDirectoriesSetting::SetGlobal(DatabaseInstance *db, DBConfig &config, const Value &input) {
	if (!config.options.enable_external_access) {
		throw InvalidInputException("Cannot change allowed_directories when enable_external_access is disabled");
	}
	if (!config.file_system) {
		throw InvalidInputException("Cannot change/set allowed_directories before the database is started");
	}
	config.options.allowed_directories.clear();
	auto &list = ListValue::GetChildren(input);
	for (auto &val : list) {
		config.AddAllowedDirectory(val.GetValue<string>());
	}
}

void AllowedDirectoriesSetting::ResetGlobal(DatabaseInstance *db, DBConfig &config) {
	if (!config.options.enable_external_access) {
		throw InvalidInputException("Cannot change allowed_directories when enable_external_access is disabled");
	}
	config.options.allowed_directories = DBConfigOptions().allowed_directories;
}

Value AllowedDirectoriesSetting::GetSetting(const ClientContext &context) {
	auto &config = DBConfig::GetConfig(context);
	vector<Value> allowed_directories;
	for (auto &dir : config.options.allowed_directories) {
		allowed_directories.emplace_back(dir);
	}
	return Value::LIST(LogicalType::VARCHAR, std::move(allowed_directories));
}

//===----------------------------------------------------------------------===//
// Allowed Paths
//===----------------------------------------------------------------------===//void
void AllowedPathsSetting::SetGlobal(DatabaseInstance *db, DBConfig &config, const Value &input) {
	if (!config.options.enable_external_access) {
		throw InvalidInputException("Cannot change allowed_paths when enable_external_access is disabled");
	}
	if (!config.file_system) {
		throw InvalidInputException("Cannot change/set allowed_paths before the database is started");
	}

	config.options.allowed_paths.clear();
	auto &list = ListValue::GetChildren(input);
	for (auto &val : list) {
		config.AddAllowedPath(val.GetValue<string>());
	}
}

void AllowedPathsSetting::ResetGlobal(DatabaseInstance *db, DBConfig &config) {
	if (!config.options.enable_external_access) {
		throw InvalidInputException("Cannot change allowed_paths when enable_external_access is disabled");
	}
	config.options.allowed_paths = DBConfigOptions().allowed_paths;
}

Value AllowedPathsSetting::GetSetting(const ClientContext &context) {
	auto &config = DBConfig::GetConfig(context);
	vector<Value> allowed_paths;
	for (auto &dir : config.options.allowed_paths) {
		allowed_paths.emplace_back(dir);
	}
	return Value::LIST(LogicalType::VARCHAR, std::move(allowed_paths));
}

//===----------------------------------------------------------------------===//
// Block Allocator Memory
//===----------------------------------------------------------------------===//
void BlockAllocatorMemorySetting::SetGlobal(DatabaseInstance *db, DBConfig &config, const Value &input) {
	const auto input_string = input.ToString();
	idx_t size;
	if (!input_string.empty() && input_string.back() == '%') {
		double percentage;
		if (!TryDoubleCast(input_string.c_str(), input_string.size() - 1, percentage, false) || percentage < 0 ||
		    percentage > 100) {
			throw InvalidInputException("Unable to parse valid percentage (input: %s)", input_string);
		}
		size = LossyNumericCast<idx_t>(percentage) * config.options.maximum_memory / 100;
	} else {
		size = DBConfig::ParseMemoryLimit(input_string);
	}
	if (db) {
		BlockAllocator::Get(*db).Resize(size);
	}
	config.options.block_allocator_size = size;
}

void BlockAllocatorMemorySetting::ResetGlobal(DatabaseInstance *db, DBConfig &config) {
	const auto size = DBConfigOptions().block_allocator_size;
	if (db) {
		BlockAllocator::Get(*db).Resize(size);
	}
	config.options.block_allocator_size = size;
}

Value BlockAllocatorMemorySetting::GetSetting(const ClientContext &context) {
	auto &config = DBConfig::GetConfig(context);
	return StringUtil::BytesToHumanReadableString(config.options.block_allocator_size);
}

//===----------------------------------------------------------------------===//
// Checkpoint Threshold
//===----------------------------------------------------------------------===//
void CheckpointThresholdSetting::SetGlobal(DatabaseInstance *db, DBConfig &config, const Value &input) {
	idx_t new_limit = DBConfig::ParseMemoryLimit(input.ToString());
	config.options.checkpoint_wal_size = new_limit;
}

Value CheckpointThresholdSetting::GetSetting(const ClientContext &context) {
	auto &config = DBConfig::GetConfig(context);
	return Value(StringUtil::BytesToHumanReadableString(config.options.checkpoint_wal_size));
}

//===----------------------------------------------------------------------===//
// Custom Profiling Settings
//===----------------------------------------------------------------------===//
bool IsEnabledOptimizer(MetricType metric, const set<OptimizerType> &disabled_optimizers) {
	auto matching_optimizer_type = MetricsUtils::GetOptimizerTypeByMetric(metric);
	if (matching_optimizer_type != OptimizerType::INVALID &&
	    disabled_optimizers.find(matching_optimizer_type) == disabled_optimizers.end()) {
		return true;
	}
	return false;
}

static profiler_settings_t FillTreeNodeSettings(unordered_map<string, string> &input,
                                                const set<OptimizerType> &disabled_optimizers) {
	profiler_settings_t metrics;

	string invalid_settings;
	for (auto &entry : input) {
		MetricType setting;
		MetricGroup group = MetricGroup::INVALID;
		try {
			setting = EnumUtil::FromString<MetricType>(StringUtil::Upper(entry.first));
		} catch (std::exception &ex) {
			try {
				group = EnumUtil::FromString<MetricGroup>(StringUtil::Upper(entry.first));
			} catch (std::exception &ex) {
				if (!invalid_settings.empty()) {
					invalid_settings += ", ";
				}
				invalid_settings += entry.first;
				continue;
			}
		}
		if (group != MetricGroup::INVALID) {
			if (entry.second == "true") {
				auto group_metrics = MetricsUtils::GetMetricsByGroupType(group);
				for (auto &metric : group_metrics) {
					if (!MetricsUtils::IsOptimizerMetric(metric) || IsEnabledOptimizer(metric, disabled_optimizers)) {
						metrics.insert(metric);
					}
				}
			}
			continue;
		}

		if (StringUtil::Lower(entry.second) == "true" &&
		    (!MetricsUtils::IsOptimizerMetric(setting) || IsEnabledOptimizer(setting, disabled_optimizers))) {
			metrics.insert(setting);
		}
	}

	if (!invalid_settings.empty()) {
		throw IOException("Invalid custom profiler settings: \"%s\"", invalid_settings);
	}
	return metrics;
}

void AddOptimizerMetrics(profiler_settings_t &settings, const set<OptimizerType> &disabled_optimizers) {
	if (settings.find(MetricType::ALL_OPTIMIZERS) != settings.end()) {
		auto optimizer_metrics = MetricsUtils::GetOptimizerMetrics();
		for (auto &metric : optimizer_metrics) {
			if (IsEnabledOptimizer(metric, disabled_optimizers)) {
				settings.insert(metric);
			}
		}
	}
}

void CustomProfilingSettingsSetting::SetLocal(ClientContext &context, const Value &input) {
	auto &config = ClientConfig::GetConfig(context);

	// parse the file content
	unordered_map<string, string> input_json;
	try {
		input_json = StringUtil::ParseJSONMap(input.ToString())->Flatten();
	} catch (std::exception &ex) {
		throw IOException("Could not parse the custom profiler settings file due to incorrect JSON: \"%s\".  Make sure "
		                  "all the keys and values start with a quote. ",
		                  input.ToString());
	}

	config.enable_profiler = true;
	auto &db_config = DBConfig::GetConfig(context);
	auto &disabled_optimizers = db_config.options.disabled_optimizers;

	auto settings = FillTreeNodeSettings(input_json, disabled_optimizers);
	AddOptimizerMetrics(settings, disabled_optimizers);
	config.profiler_settings = settings;
}

void CustomProfilingSettingsSetting::ResetLocal(ClientContext &context) {
	auto &config = ClientConfig::GetConfig(context);
	config.enable_profiler = ClientConfig().enable_profiler;
	config.profiler_settings = MetricsUtils::GetDefaultMetrics();
}

Value CustomProfilingSettingsSetting::GetSetting(const ClientContext &context) {
	auto &config = ClientConfig::GetConfig(context);

	string profiling_settings_str;
	for (auto &entry : config.profiler_settings) {
		if (!profiling_settings_str.empty()) {
			profiling_settings_str += ", ";
		}
		profiling_settings_str += StringUtil::Format("\"%s\": \"true\"", EnumUtil::ToString(entry));
	}
	return Value(StringUtil::Format("{%s}", profiling_settings_str));
}

//===----------------------------------------------------------------------===//
// Custom User Agent
//===----------------------------------------------------------------------===//
void CustomUserAgentSetting::SetGlobal(DatabaseInstance *db, DBConfig &config, const Value &input) {
	auto new_value = input.GetValue<string>();
	if (db) {
		throw InvalidInputException("Cannot change custom_user_agent setting while database is running");
	}
	config.options.custom_user_agent =
	    config.options.custom_user_agent.empty() ? new_value : config.options.custom_user_agent + " " + new_value;
}

void CustomUserAgentSetting::ResetGlobal(DatabaseInstance *db, DBConfig &config) {
	if (db) {
		throw InvalidInputException("Cannot change custom_user_agent setting while database is running");
	}
	config.options.custom_user_agent = DBConfigOptions().custom_user_agent;
}

//===----------------------------------------------------------------------===//
// Default Block Size
//===----------------------------------------------------------------------===//
void DefaultBlockSizeSetting::SetGlobal(DatabaseInstance *db, DBConfig &config, const Value &input) {
	auto block_alloc_size = input.GetValue<uint64_t>();
	Storage::VerifyBlockAllocSize(block_alloc_size);
	config.options.default_block_alloc_size = block_alloc_size;
}

void DefaultBlockSizeSetting::ResetGlobal(DatabaseInstance *db, DBConfig &config) {
	config.options.default_block_alloc_size = DBConfigOptions().default_block_alloc_size;
}

Value DefaultBlockSizeSetting::GetSetting(const ClientContext &context) {
	auto &config = DBConfig::GetConfig(context);
	return Value::UBIGINT(config.options.default_block_alloc_size);
}

//===----------------------------------------------------------------------===//
// Default Collation
//===----------------------------------------------------------------------===//
void DefaultCollationSetting::OnSet(SettingCallbackInfo &info, Value &input) {
	if (info.context) {
		ExpressionBinder::TestCollation(*info.context, input.ToString());
	}
}

//===----------------------------------------------------------------------===//
// Default Null Order
//===----------------------------------------------------------------------===//
void DefaultNullOrderSetting::OnSet(SettingCallbackInfo &, Value &input) {
	auto parameter = StringUtil::Lower(input.ToString());

	if (parameter == "nulls_first" || parameter == "nulls first" || parameter == "null first" || parameter == "first") {
		input = Value("NULLS_FIRST");
	} else if (parameter == "nulls_last" || parameter == "nulls last" || parameter == "null last" ||
	           parameter == "last") {
		input = Value("NULLS_LAST");
	} else if (parameter == "nulls_first_on_asc_last_on_desc" || parameter == "sqlite" || parameter == "mysql") {
		input = Value("NULLS_FIRST_ON_ASC_LAST_ON_DESC");
	} else if (parameter == "nulls_last_on_asc_first_on_desc" || parameter == "postgres") {
		input = Value("NULLS_LAST_ON_ASC_FIRST_ON_DESC");
	} else {
		throw ParserException("Unrecognized parameter for option NULL_ORDER \"%s\", expected either NULLS FIRST, NULLS "
		                      "LAST, SQLite, MySQL or Postgres",
		                      parameter);
	}
}

//===----------------------------------------------------------------------===//
// Default Order
//===----------------------------------------------------------------------===//
void DefaultOrderSetting::OnSet(SettingCallbackInfo &, Value &input) {
	auto parameter = StringUtil::Lower(input.ToString());
	if (parameter == "ascending" || parameter == "asc") {
		input = Value("ASC");
	} else if (parameter == "descending" || parameter == "desc") {
		input = Value("DESC");
	} else {
		throw InvalidInputException("Unrecognized parameter for option DEFAULT_ORDER \"%s\". Expected ASC or DESC.",
		                            parameter);
	}
}

//===----------------------------------------------------------------------===//
// Default Secret Storage
//===----------------------------------------------------------------------===//
void DefaultSecretStorageSetting::SetGlobal(DatabaseInstance *db, DBConfig &config, const Value &input) {
	config.secret_manager->SetDefaultStorage(input.ToString());
}

void DefaultSecretStorageSetting::ResetGlobal(DatabaseInstance *db, DBConfig &config) {
	config.secret_manager->ResetDefaultStorage();
}

Value DefaultSecretStorageSetting::GetSetting(const ClientContext &context) {
	auto &config = DBConfig::GetConfig(context);
	return config.secret_manager->DefaultStorage();
}

//===----------------------------------------------------------------------===//
// Disabled Compression Methods
//===----------------------------------------------------------------------===//
void DisabledCompressionMethodsSetting::SetGlobal(DatabaseInstance *db, DBConfig &config, const Value &input) {
	auto list = StringUtil::Split(input.ToString(), ",");
	vector<CompressionType> disabled_compression_methods;
	for (auto &entry : list) {
		auto param = StringUtil::Lower(entry);
		StringUtil::Trim(param);
		if (param.empty()) {
			continue;
		}
		if (param == "none") {
			disabled_compression_methods.clear();
			break;
		}
		auto compression_type = CompressionTypeFromString(param);
		if (compression_type == CompressionType::COMPRESSION_UNCOMPRESSED) {
			throw InvalidInputException("Uncompressed compression cannot be disabled");
		}
		if (compression_type == CompressionType::COMPRESSION_AUTO) {
			throw InvalidInputException("Unrecognized compression method \"%s\"", entry);
		}
		disabled_compression_methods.push_back(compression_type);
	}
	config.SetDisabledCompressionMethods(disabled_compression_methods);
}

void DisabledCompressionMethodsSetting::ResetGlobal(DatabaseInstance *db, DBConfig &config) {
	vector<CompressionType> disabled_compression_methods;
	config.SetDisabledCompressionMethods(disabled_compression_methods);
}

Value DisabledCompressionMethodsSetting::GetSetting(const ClientContext &context) {
	auto &config = DBConfig::GetConfig(context);
	string result;
	for (auto &optimizer : config.GetDisabledCompressionMethods()) {
		if (!result.empty()) {
			result += ",";
		}
		result += CompressionTypeToString(optimizer);
	}
	return Value(result);
}

//===----------------------------------------------------------------------===//
// Disabled Filesystems
//===----------------------------------------------------------------------===//
void DisabledFilesystemsSetting::SetGlobal(DatabaseInstance *db_p, DBConfig &config, const Value &input) {
	auto &db = GetDB<DisabledFilesystemsSetting>(db_p);
	auto &fs = FileSystem::GetFileSystem(db);
	auto list = StringUtil::Split(input.ToString(), ",");
	fs.SetDisabledFileSystems(list);
}

void DisabledFilesystemsSetting::ResetGlobal(DatabaseInstance *db_p, DBConfig &config) {
	auto &db = GetDB<DisabledFilesystemsSetting>(db_p);
	auto &fs = FileSystem::GetFileSystem(db);
	fs.SetDisabledFileSystems(vector<string>());
}

Value DisabledFilesystemsSetting::GetSetting(const ClientContext &context) {
	return Value("");
}

//===----------------------------------------------------------------------===//
// Disabled Optimizers
//===----------------------------------------------------------------------===//
void DisabledOptimizersSetting::SetGlobal(DatabaseInstance *db, DBConfig &config, const Value &input) {
	auto list = StringUtil::Split(input.ToString(), ",");
	set<OptimizerType> disabled_optimizers;
	for (auto &entry : list) {
		auto param = StringUtil::Lower(entry);
		StringUtil::Trim(param);
		if (param.empty()) {
			continue;
		}
		disabled_optimizers.insert(OptimizerTypeFromString(param));
	}
	config.options.disabled_optimizers = std::move(disabled_optimizers);
}

void DisabledOptimizersSetting::ResetGlobal(DatabaseInstance *db, DBConfig &config) {
	config.options.disabled_optimizers = DBConfigOptions().disabled_optimizers;
}

Value DisabledOptimizersSetting::GetSetting(const ClientContext &context) {
	auto &config = DBConfig::GetConfig(context);
	string result;
	for (auto &optimizer : config.options.disabled_optimizers) {
		if (!result.empty()) {
			result += ",";
		}
		result += OptimizerTypeToString(optimizer);
	}
	return Value(result);
}

//===----------------------------------------------------------------------===//
// Duckdb Api
//===----------------------------------------------------------------------===//
void DuckDBAPISetting::SetGlobal(DatabaseInstance *db, DBConfig &config, const Value &input) {
	auto new_value = input.GetValue<string>();
	if (db) {
		throw InvalidInputException("Cannot change duckdb_api setting while database is running");
	}
	config.options.duckdb_api = new_value;
}

void DuckDBAPISetting::ResetGlobal(DatabaseInstance *db, DBConfig &config) {
	if (db) {
		throw InvalidInputException("Cannot change duckdb_api setting while database is running");
	}
	config.options.duckdb_api = GetDefaultUserAgent();
}

Value DuckDBAPISetting::GetSetting(const ClientContext &context) {
	auto &config = DBConfig::GetConfig(context);
	return Value(config.options.duckdb_api);
}

//===----------------------------------------------------------------------===//
// Enable External Access
//===----------------------------------------------------------------------===//
bool EnableExternalAccessSetting::OnGlobalSet(DatabaseInstance *db, DBConfig &config, const Value &input) {
	if (!db) {
		return true;
	}
	if (input.GetValue<bool>()) {
		throw InvalidInputException("Cannot change enable_external_access setting while database is running");
	}
	if (db && config.options.enable_external_access) {
		// we are turning off external access - add any already attached databases to the list of accepted paths
		auto &db_manager = DatabaseManager::Get(*db);
		auto attached_paths = db_manager.GetAttachedDatabasePaths();
		for (auto &path : attached_paths) {
			config.AddAllowedPath(path);
			config.AddAllowedPath(path + ".wal");
			config.AddAllowedPath(path + ".checkpoint.wal");
			config.AddAllowedPath(path + ".recovery.wal");
		}
	}
	if (config.options.use_temporary_directory && !config.options.temporary_directory.empty()) {
		// if temp directory is enabled we can also write there
		config.AddAllowedDirectory(config.options.temporary_directory);
	}
	return true;
}

bool EnableExternalAccessSetting::OnGlobalReset(DatabaseInstance *db, DBConfig &config) {
	if (db) {
		throw InvalidInputException("Cannot change enable_external_access setting while database is running");
	}
	return true;
}

//===----------------------------------------------------------------------===//
// Enable External File Cache
//===----------------------------------------------------------------------===//
void EnableExternalFileCacheSetting::SetGlobal(DatabaseInstance *db, DBConfig &config, const Value &input) {
	config.options.enable_external_file_cache = input.GetValue<bool>();
	if (db) {
		ExternalFileCache::Get(*db).SetEnabled(config.options.enable_external_file_cache);
	}
}

void EnableExternalFileCacheSetting::ResetGlobal(DatabaseInstance *db, DBConfig &config) {
	config.options.enable_external_file_cache = DBConfigOptions().enable_external_file_cache;
	if (db) {
		ExternalFileCache::Get(*db).SetEnabled(config.options.enable_external_file_cache);
	}
}

Value EnableExternalFileCacheSetting::GetSetting(const ClientContext &context) {
	auto &config = DBConfig::GetConfig(context);
	return Value(config.options.enable_external_file_cache);
}

//===----------------------------------------------------------------------===//
// Enable Logging
//===----------------------------------------------------------------------===//
Value EnableLogging::GetSetting(const ClientContext &context) {
	return context.db->GetLogManager().GetConfig().enabled;
}
void EnableLogging::SetGlobal(DatabaseInstance *db_p, DBConfig &config, const Value &parameter) {
	auto &db = GetDB<EnableLogging>(db_p);
	db.GetLogManager().SetEnableLogging(parameter.GetValue<bool>());
}

void EnableLogging::ResetGlobal(DatabaseInstance *db_p, DBConfig &config) {
	auto &db = GetDB<EnableLogging>(db_p);
	db.GetLogManager().SetEnableLogging(false);
}

//===----------------------------------------------------------------------===//
// Force VARIANT Shredding
//===----------------------------------------------------------------------===//

void ForceVariantShredding::SetGlobal(DatabaseInstance *_, DBConfig &config, const Value &value) {
	auto &force_variant_shredding = config.options.force_variant_shredding;

	if (value.type().id() != LogicalTypeId::VARCHAR) {
		throw InvalidInputException("The argument to 'force_variant_shredding' should be of type VARCHAR, not %s",
		                            value.type().ToString());
	}

	auto logical_type = TransformStringToLogicalType(value.GetValue<string>());
	TypeVisitor::Contains(logical_type, [](const LogicalType &type) {
		if (type.IsNested()) {
			if (type.id() != LogicalTypeId::STRUCT && type.id() != LogicalTypeId::LIST) {
				throw InvalidInputException("Shredding can consist of the nested types LIST (for ARRAY Variant values) "
				                            "or STRUCT (for OBJECT Variant values), not %s",
				                            type.ToString());
			}
			if (type.id() == LogicalTypeId::STRUCT && StructType::IsUnnamed(type)) {
				throw InvalidInputException("STRUCT types in the shredding can not be empty");
			}
			return false;
		}
		switch (type.id()) {
		case LogicalTypeId::BOOLEAN:
		case LogicalTypeId::TINYINT:
		case LogicalTypeId::SMALLINT:
		case LogicalTypeId::INTEGER:
		case LogicalTypeId::BIGINT:
		case LogicalTypeId::HUGEINT:
		case LogicalTypeId::UTINYINT:
		case LogicalTypeId::USMALLINT:
		case LogicalTypeId::UINTEGER:
		case LogicalTypeId::UBIGINT:
		case LogicalTypeId::UHUGEINT:
		case LogicalTypeId::FLOAT:
		case LogicalTypeId::DOUBLE:
		case LogicalTypeId::DECIMAL:
		case LogicalTypeId::DATE:
		case LogicalTypeId::TIME:
		case LogicalTypeId::TIME_TZ:
		case LogicalTypeId::TIMESTAMP_TZ:
		case LogicalTypeId::TIMESTAMP:
		case LogicalTypeId::TIMESTAMP_SEC:
		case LogicalTypeId::TIMESTAMP_MS:
		case LogicalTypeId::TIMESTAMP_NS:
		case LogicalTypeId::BLOB:
		case LogicalTypeId::VARCHAR:
		case LogicalTypeId::UUID:
		case LogicalTypeId::BIGNUM:
		case LogicalTypeId::TIME_NS:
		case LogicalTypeId::INTERVAL:
		case LogicalTypeId::BIT:
		case LogicalTypeId::GEOMETRY:
			break;
		default:
			throw InvalidInputException("Variants can not be shredded on type: %s", type.ToString());
		}
		return false;
	});

	auto shredding_type = TypeVisitor::VisitReplace(logical_type, [](const LogicalType &type) {
		return LogicalType::STRUCT({{"untyped_value_index", LogicalType::UINTEGER}, {"typed_value", type}});
	});
	force_variant_shredding =
	    LogicalType::STRUCT({{"unshredded", VariantShredding::GetUnshreddedType()}, {"shredded", shredding_type}});
}

void ForceVariantShredding::ResetGlobal(DatabaseInstance *db, DBConfig &config) {
	config.options.force_variant_shredding = LogicalType::INVALID;
}

Value ForceVariantShredding::GetSetting(const ClientContext &context) {
	auto &config = DBConfig::GetConfig(context);
	return Value(config.options.force_variant_shredding.ToString());
}

//===----------------------------------------------------------------------===//
// Extension Directory
//===----------------------------------------------------------------------===//
void ExtensionDirectoriesSetting::SetGlobal(DatabaseInstance *db, DBConfig &config, const Value &input) {
	config.options.extension_directories.clear();

	auto &list = ListValue::GetChildren(input);
	for (auto &val : list) {
		config.options.extension_directories.emplace_back(val.GetValue<string>());
	}
}

void ExtensionDirectoriesSetting::ResetGlobal(DatabaseInstance *db, DBConfig &config) {
	config.options.extension_directories = DBConfigOptions().extension_directories;
}

Value ExtensionDirectoriesSetting::GetSetting(const ClientContext &context) {
	auto &config = DBConfig::GetConfig(context);
	vector<Value> extension_directories;
	for (auto &dir : config.options.extension_directories) {
		extension_directories.emplace_back(dir);
	}
	return Value::LIST(LogicalType::VARCHAR, std::move(extension_directories));
}

//===----------------------------------------------------------------------===//
// Logging Mode
//===----------------------------------------------------------------------===//
Value LoggingMode::GetSetting(const ClientContext &context) {
	return EnumUtil::ToString(context.db->GetLogManager().GetConfig().mode);
}
void LoggingMode::SetGlobal(DatabaseInstance *db_p, DBConfig &config, const Value &parameter) {
	auto &db = GetDB<LoggingMode>(db_p);
	db.GetLogManager().SetLogMode(EnumUtil::FromString<LogMode>(parameter.GetValue<string>()));
}

void LoggingMode::ResetGlobal(DatabaseInstance *db_p, DBConfig &config) {
	auto &db = GetDB<LoggingMode>(db_p);
	db.GetLogManager().SetLogMode(LogMode::LEVEL_ONLY);
}

//===----------------------------------------------------------------------===//
// Logging Level
//===----------------------------------------------------------------------===//
Value LoggingLevel::GetSetting(const ClientContext &context) {
	return EnumUtil::ToString(context.db->GetLogManager().GetConfig().level);
}
void LoggingLevel::SetGlobal(DatabaseInstance *db_p, DBConfig &config, const Value &parameter) {
	auto &db = GetDB<LoggingLevel>(db_p);
	db.GetLogManager().SetLogLevel(EnumUtil::FromString<LogLevel>(parameter.GetValue<string>()));
}

void LoggingLevel::ResetGlobal(DatabaseInstance *db_p, DBConfig &config) {
	auto &db = GetDB<LoggingLevel>(db_p);
	db.GetLogManager().SetLogLevel(LogConfig::DEFAULT_LOG_LEVEL);
}

//===----------------------------------------------------------------------===//
// Logging Storage
//===----------------------------------------------------------------------===//
Value LoggingStorage::GetSetting(const ClientContext &context) {
	return context.db->GetLogManager().GetConfig().storage;
}
void LoggingStorage::SetGlobal(DatabaseInstance *db_p, DBConfig &config, const Value &parameter) {
	auto &db = GetDB<LoggingStorage>(db_p);
	db.GetLogManager().SetLogStorage(db, parameter.GetValue<string>());
}

void LoggingStorage::ResetGlobal(DatabaseInstance *db_p, DBConfig &config) {
	auto &db = GetDB<LoggingStorage>(db_p);
	db.GetLogManager().SetLogStorage(db, LogConfig::DEFAULT_LOG_STORAGE);
}

//===----------------------------------------------------------------------===//
// Enabled Loggers
//===----------------------------------------------------------------------===//
Value EnabledLogTypes::GetSetting(const ClientContext &context) {
	vector<string> loggers;
	for (const auto &item : context.db->GetLogManager().GetConfig().enabled_log_types) {
		loggers.push_back(item);
	}
	return StringUtil::Join(loggers, ",");
}
void EnabledLogTypes::SetGlobal(DatabaseInstance *db_p, DBConfig &config, const Value &parameter) {
	auto &db = GetDB<EnabledLogTypes>(db_p);

	auto values = StringUtil::Split(parameter.GetValue<string>(), ",");
	unordered_set<string> set;
	for (const auto &value : values) {
		set.insert(value);
	}
	db.GetLogManager().SetEnabledLogTypes(set);
}

void EnabledLogTypes::ResetGlobal(DatabaseInstance *db_p, DBConfig &config) {
	auto &db = GetDB<EnabledLogTypes>(db_p);
	unordered_set<string> set;
	db.GetLogManager().SetEnabledLogTypes(set);
}

//===----------------------------------------------------------------------===//
// Disabled Loggers
//===----------------------------------------------------------------------===//
Value DisabledLogTypes::GetSetting(const ClientContext &context) {
	vector<string> loggers;
	for (const auto &item : context.db->GetLogManager().GetConfig().disabled_log_types) {
		loggers.push_back(item);
	}
	return StringUtil::Join(loggers, ",");
}
void DisabledLogTypes::SetGlobal(DatabaseInstance *db_p, DBConfig &config, const Value &parameter) {
	auto &db = GetDB<DisabledLogTypes>(db_p);
	auto values = StringUtil::Split(parameter.GetValue<string>(), ",");
	unordered_set<string> set;
	for (const auto &value : values) {
		set.insert(value);
	}
	db.GetLogManager().SetDisabledLogTypes(set);
}

void DisabledLogTypes::ResetGlobal(DatabaseInstance *db_p, DBConfig &config) {
	auto &db = GetDB<DisabledLogTypes>(db_p);
	unordered_set<string> set;
	db.GetLogManager().SetDisabledLogTypes(set);
}

//===----------------------------------------------------------------------===//
// Enable Profiling
//===----------------------------------------------------------------------===//
void EnableProfilingSetting::SetLocal(ClientContext &context, const Value &input) {
	auto parameter = StringUtil::Lower(input.ToString());

	auto &config = ClientConfig::GetConfig(context);
	config.enable_profiler = true;
	config.emit_profiler_output = true;

	if (parameter == "json") {
		config.profiler_print_format = ProfilerPrintFormat::JSON;
	} else if (parameter == "query_tree") {
		config.profiler_print_format = ProfilerPrintFormat::QUERY_TREE;
	} else if (parameter == "query_tree_optimizer") {
		config.profiler_print_format = ProfilerPrintFormat::QUERY_TREE_OPTIMIZER;

		// add optimizer settings to the profiler settings
		auto optimizer_settings = MetricsUtils::GetOptimizerMetrics();
		for (auto &setting : optimizer_settings) {
			config.profiler_settings.insert(setting);
		}

		// add the phase timing settings to the profiler settings
		auto phase_timing_settings = MetricsUtils::GetPhaseTimingMetrics();
		for (auto &setting : phase_timing_settings) {
			config.profiler_settings.insert(setting);
		}
	} else if (parameter == "no_output") {
		config.profiler_print_format = ProfilerPrintFormat::NO_OUTPUT;
		config.emit_profiler_output = false;
	} else if (parameter == "html") {
		config.profiler_print_format = ProfilerPrintFormat::HTML;
	} else if (parameter == "graphviz") {
		config.profiler_print_format = ProfilerPrintFormat::GRAPHVIZ;
	} else {
		throw ParserException(
		    "Unrecognized print format %s, supported formats: [json, query_tree, query_tree_optimizer, no_output]",
		    parameter);
	}
}

void EnableProfilingSetting::ResetLocal(ClientContext &context) {
	auto &config = ClientConfig::GetConfig(context);
	config.profiler_print_format = ClientConfig().profiler_print_format;
	config.enable_profiler = ClientConfig().enable_profiler;
	config.emit_profiler_output = ClientConfig().emit_profiler_output;
	config.profiler_settings = ClientConfig().profiler_settings;
}

Value EnableProfilingSetting::GetSetting(const ClientContext &context) {
	auto &config = ClientConfig::GetConfig(context);
	if (!config.enable_profiler) {
		return Value();
	}
	switch (config.profiler_print_format) {
	case ProfilerPrintFormat::JSON:
		return Value("json");
	case ProfilerPrintFormat::QUERY_TREE:
		return Value("query_tree");
	case ProfilerPrintFormat::QUERY_TREE_OPTIMIZER:
		return Value("query_tree_optimizer");
	case ProfilerPrintFormat::NO_OUTPUT:
		return Value("no_output");
	case ProfilerPrintFormat::HTML:
		return Value("html");
	case ProfilerPrintFormat::GRAPHVIZ:
		return Value("graphviz");
	default:
		throw InternalException("Unsupported profiler print format");
	}
}

//===----------------------------------------------------------------------===//
// Enable Progress Bar Print
//===----------------------------------------------------------------------===//
void EnableProgressBarPrintSetting::SetLocal(ClientContext &context, const Value &input) {
	auto &config = ClientConfig::GetConfig(context);
	ProgressBar::SystemOverrideCheck(config);
	config.print_progress_bar = input.GetValue<bool>();
}

void EnableProgressBarPrintSetting::ResetLocal(ClientContext &context) {
	auto &config = ClientConfig::GetConfig(context);
	ProgressBar::SystemOverrideCheck(config);
	config.print_progress_bar = ClientConfig().print_progress_bar;
}

Value EnableProgressBarPrintSetting::GetSetting(const ClientContext &context) {
	return Value::BOOLEAN(ClientConfig::GetConfig(context).print_progress_bar);
}

//===----------------------------------------------------------------------===//
// Enable Progress Bar
//===----------------------------------------------------------------------===//
bool EnableProgressBarSetting::OnLocalSet(ClientContext &context, const Value &input) {
	auto &config = ClientConfig::GetConfig(context);
	ProgressBar::SystemOverrideCheck(config);
	return true;
}

bool EnableProgressBarSetting::OnLocalReset(ClientContext &context) {
	auto &config = ClientConfig::GetConfig(context);
	ProgressBar::SystemOverrideCheck(config);
	return true;
}

//===----------------------------------------------------------------------===//
// External Threads
//===----------------------------------------------------------------------===//
bool ExternalThreadsSetting::OnGlobalSet(DatabaseInstance *db, DBConfig &config, const Value &input) {
	auto new_val = input.GetValue<int64_t>();
	if (new_val < 0) {
		throw SyntaxException("Must have a non-negative number of external threads!");
	}
	auto new_external_threads = NumericCast<idx_t>(new_val);
	if (db) {
		TaskScheduler::GetScheduler(*db).SetThreads(config.options.maximum_threads, new_external_threads);
	}
	return true;
}

bool ExternalThreadsSetting::OnGlobalReset(DatabaseInstance *db, DBConfig &config) {
	idx_t new_external_threads = DBConfigOptions().external_threads;
	if (db) {
		TaskScheduler::GetScheduler(*db).SetThreads(config.options.maximum_threads, new_external_threads);
	}
	return true;
}

//===----------------------------------------------------------------------===//
// File Search Path
//===----------------------------------------------------------------------===//
void FileSearchPathSetting::SetLocal(ClientContext &context, const Value &input) {
	auto parameter = input.ToString();
	auto &client_data = ClientData::Get(context);
	client_data.file_search_path = parameter;
}

void FileSearchPathSetting::ResetLocal(ClientContext &context) {
	auto &client_data = ClientData::Get(context);
	client_data.file_search_path.clear();
}

Value FileSearchPathSetting::GetSetting(const ClientContext &context) {
	auto &client_data = ClientData::Get(context);
	return Value(client_data.file_search_path);
}

//===----------------------------------------------------------------------===//
// Force Bitpacking Mode
//===----------------------------------------------------------------------===//
void ForceBitpackingModeSetting::SetGlobal(DatabaseInstance *db, DBConfig &config, const Value &input) {
	auto mode_str = StringUtil::Lower(input.ToString());
	auto mode = BitpackingModeFromString(mode_str);
	if (mode == BitpackingMode::INVALID) {
		throw ParserException("Unrecognized option for force_bitpacking_mode, expected none, constant, constant_delta, "
		                      "delta_for, or for");
	}
	config.options.force_bitpacking_mode = mode;
}

void ForceBitpackingModeSetting::ResetGlobal(DatabaseInstance *db, DBConfig &config) {
	config.options.force_bitpacking_mode = DBConfigOptions().force_bitpacking_mode;
}

Value ForceBitpackingModeSetting::GetSetting(const ClientContext &context) {
	return Value(BitpackingModeToString(context.db->config.options.force_bitpacking_mode));
}

//===----------------------------------------------------------------------===//
// Force Compression
//===----------------------------------------------------------------------===//
void ForceCompressionSetting::SetGlobal(DatabaseInstance *db, DBConfig &config, const Value &input) {
	auto compression = StringUtil::Lower(input.ToString());
	if (compression == "none" || compression == "auto") {
		config.options.force_compression = CompressionType::COMPRESSION_AUTO;
	} else {
		auto compression_type = CompressionTypeFromString(compression);
		//! FIXME: do we want to try to retrieve the AttachedDatabase here to get the StorageManager ??
		auto compression_availability_result = CompressionTypeIsAvailable(compression_type);
		if (!compression_availability_result.IsAvailable()) {
			if (compression_availability_result.IsDeprecated()) {
				throw ParserException("Attempted to force a deprecated compression type (%s)",
				                      CompressionTypeToString(compression_type));
			} else {
				throw ParserException("Attempted to force a compression type that isn't available yet (%s)",
				                      CompressionTypeToString(compression_type));
			}
		}
		if (compression_type == CompressionType::COMPRESSION_AUTO) {
			auto compression_types = StringUtil::Join(ListCompressionTypes(), ", ");
			throw ParserException("Unrecognized option for PRAGMA force_compression, expected %s", compression_types);
		}
		config.options.force_compression = compression_type;
	}
}

void ForceCompressionSetting::ResetGlobal(DatabaseInstance *db, DBConfig &config) {
	config.options.force_compression = DBConfigOptions().force_compression;
}

Value ForceCompressionSetting::GetSetting(const ClientContext &context) {
	auto &config = DBConfig::GetConfig(*context.db);
	return CompressionTypeToString(config.options.force_compression);
}

//===----------------------------------------------------------------------===//
// Home Directory
//===----------------------------------------------------------------------===//
void HomeDirectorySetting::SetLocal(ClientContext &context, const Value &input) {
	auto &config = ClientConfig::GetConfig(context);
	if (!input.IsNull() && FileSystem::GetFileSystem(context).IsRemoteFile(input.ToString())) {
		throw InvalidInputException("Cannot set the home directory to a remote path");
	}
	config.home_directory = input.IsNull() ? string() : input.ToString();
}

//===----------------------------------------------------------------------===//
// Enable H T T P Logging
//===----------------------------------------------------------------------===//
void EnableHTTPLoggingSetting::SetLocal(ClientContext &context, const Value &input) {
	auto &config = ClientConfig::GetConfig(context);
	config.enable_http_logging = input.GetValue<bool>();

	// NOTE: this is a deprecated setting: we mimick the old behaviour by setting the log storage output to STDOUT and
	// enabling logging for http only. Note that this behaviour is slightly wonky in that it sets all sorts of logging
	// config
	auto &log_manager = LogManager::Get(context);
	if (config.enable_http_logging) {
		log_manager.SetEnableLogging(true);
		log_manager.SetLogLevel(HTTPLogType::LEVEL);
		unordered_set<string> enabled_log_types = {HTTPLogType::NAME};
		log_manager.SetEnabledLogTypes(enabled_log_types);
		log_manager.SetLogStorage(*context.db, LogConfig::STDOUT_STORAGE_NAME);
	} else {
		log_manager.SetEnableLogging(false);
	}
}

void EnableHTTPLoggingSetting::ResetLocal(ClientContext &context) {
	ClientConfig::GetConfig(context).enable_http_logging = ClientConfig().enable_http_logging;
}

Value EnableHTTPLoggingSetting::GetSetting(const ClientContext &context) {
	auto &config = ClientConfig::GetConfig(context);
	return Value::BOOLEAN(config.enable_http_logging);
}

//===----------------------------------------------------------------------===//
// H T T P Logging Output
//===----------------------------------------------------------------------===//
void HTTPLoggingOutputSetting::SetLocal(ClientContext &context, const Value &input) {
	throw NotImplementedException("This setting is deprecated and can no longer be used. Check out the DuckDB docs on "
	                              "logging for more information");
}

void HTTPLoggingOutputSetting::ResetLocal(ClientContext &context) {
	throw NotImplementedException("This setting is deprecated and can no longer be used. Check out the DuckDB docs on "
	                              "logging for more information");
}

Value HTTPLoggingOutputSetting::GetSetting(const ClientContext &context) {
	auto &config = ClientConfig::GetConfig(context);
	return Value(config.http_logging_output);
}

//===----------------------------------------------------------------------===//
// Index Scan Percentage
//===----------------------------------------------------------------------===//
void IndexScanPercentageSetting::OnSet(SettingCallbackInfo &, Value &input) {
	auto index_scan_percentage = input.GetValue<double>();
	if (index_scan_percentage < 0 || index_scan_percentage > 1.0) {
		throw InvalidInputException("the index scan percentage must be within [0, 1]");
	}
}

//===----------------------------------------------------------------------===//
// Lambda Syntax Setting
//===----------------------------------------------------------------------===//
void LambdaSyntaxSetting::SetLocal(ClientContext &context, const Value &input) {
	auto setting_type = EnumUtil::FromString<LambdaSyntax>(input.ToString());
	auto &config = ClientConfig::GetConfig(context);
	config.lambda_syntax = setting_type;
}

void LambdaSyntaxSetting::ResetLocal(ClientContext &context) {
	auto &config = ClientConfig::GetConfig(context);
	config.lambda_syntax = LambdaSyntax::DEFAULT;
}

Value LambdaSyntaxSetting::GetSetting(const ClientContext &context) {
	const auto &config = ClientConfig::GetConfig(context);
	return Value(EnumUtil::ToString(config.lambda_syntax));
}

//===----------------------------------------------------------------------===//
// Log Query Path
//===----------------------------------------------------------------------===//
void LogQueryPathSetting::SetLocal(ClientContext &context, const Value &input) {
	auto &client_data = ClientData::Get(context);
	auto path = input.ToString();
	if (path.empty()) {
		// empty path: clean up query writer
		client_data.log_query_writer = nullptr;
	} else {
		client_data.log_query_writer = make_uniq<BufferedFileWriter>(FileSystem::GetFileSystem(context), path,
		                                                             BufferedFileWriter::DEFAULT_OPEN_FLAGS);
	}
}

void LogQueryPathSetting::ResetLocal(ClientContext &context) {
	auto &client_data = ClientData::Get(context);
	client_data.log_query_writer = nullptr;
}

Value LogQueryPathSetting::GetSetting(const ClientContext &context) {
	auto &client_data = ClientData::Get(context);
	return client_data.log_query_writer ? Value(client_data.log_query_writer->path) : Value();
}

//===----------------------------------------------------------------------===//
// Max Memory
//===----------------------------------------------------------------------===//
void MaxMemorySetting::SetGlobal(DatabaseInstance *db, DBConfig &config, const Value &input) {
	config.options.maximum_memory = DBConfig::ParseMemoryLimit(input.ToString());
	if (db) {
		BufferManager::GetBufferManager(*db).SetMemoryLimit(config.options.maximum_memory);
	}
}

void MaxMemorySetting::ResetGlobal(DatabaseInstance *db, DBConfig &config) {
	config.SetDefaultMaxMemory();
}

Value MaxMemorySetting::GetSetting(const ClientContext &context) {
	auto &config = DBConfig::GetConfig(context);
	return Value(StringUtil::BytesToHumanReadableString(config.options.maximum_memory));
}

//===----------------------------------------------------------------------===//
// Max Temp Directory Size
//===----------------------------------------------------------------------===//
void MaxTempDirectorySizeSetting::SetGlobal(DatabaseInstance *db, DBConfig &config, const Value &input) {
	if (input == "90% of available disk space") {
		ResetGlobal(db, config);
		return;
	}
	auto maximum_swap_space = DBConfig::ParseMemoryLimit(input.ToString());
	if (maximum_swap_space == DConstants::INVALID_INDEX) {
		// We use INVALID_INDEX to indicate that the value is not set by the user
		// use one lower to indicate 'unlimited'
		maximum_swap_space--;
	}
	if (!db) {
		config.options.maximum_swap_space = maximum_swap_space;
		return;
	}
	auto &buffer_manager = BufferManager::GetBufferManager(*db);
	buffer_manager.SetSwapLimit(maximum_swap_space);
	config.options.maximum_swap_space = maximum_swap_space;
}

void MaxTempDirectorySizeSetting::ResetGlobal(DatabaseInstance *db, DBConfig &config) {
	config.options.maximum_swap_space = DConstants::INVALID_INDEX;
	if (!db) {
		return;
	}
	auto &buffer_manager = BufferManager::GetBufferManager(*db);
	buffer_manager.SetSwapLimit();
}

Value MaxTempDirectorySizeSetting::GetSetting(const ClientContext &context) {
	auto &config = DBConfig::GetConfig(context);
	if (config.options.maximum_swap_space != DConstants::INVALID_INDEX) {
		// Explicitly set by the user
		return Value(StringUtil::BytesToHumanReadableString(config.options.maximum_swap_space));
	}
	auto &buffer_manager = BufferManager::GetBufferManager(context);
	// Database is initialized, use the setting from the temporary directory
	auto max_swap = buffer_manager.GetMaxSwap();
	if (max_swap.IsValid()) {
		return Value(StringUtil::BytesToHumanReadableString(max_swap.GetIndex()));
	} else {
		// The temp directory has not been used yet
		return Value("90% of available disk space");
	}
}

//===----------------------------------------------------------------------===//
// Ordered Aggregate Threshold
//===----------------------------------------------------------------------===//
void OrderedAggregateThresholdSetting::OnSet(SettingCallbackInfo &info, Value &input) {
	const auto param = input.GetValue<uint64_t>();
	if (param <= 0) {
		throw ParserException("Invalid option for PRAGMA ordered_aggregate_threshold, value must be positive");
	}
}

//===----------------------------------------------------------------------===//
// Password
//===----------------------------------------------------------------------===//
void PasswordSetting::SetGlobal(DatabaseInstance *db, DBConfig &config, const Value &input) {
	// nop
}

void PasswordSetting::ResetGlobal(DatabaseInstance *db, DBConfig &config) {
	// nop
}

Value PasswordSetting::GetSetting(const ClientContext &context) {
	return Value();
}

//===----------------------------------------------------------------------===//
// Perfect Ht Threshold
//===----------------------------------------------------------------------===//
void PerfectHtThresholdSetting::OnSet(SettingCallbackInfo &info, Value &input) {
	auto bits = input.GetValue<int64_t>();
	if (bits < 0 || bits > 32) {
		throw ParserException("Perfect HT threshold out of range: should be within range 0 - 32");
	}
}

//===----------------------------------------------------------------------===//
// Profile Output
//===----------------------------------------------------------------------===//
void ProfileOutputSetting::SetLocal(ClientContext &context, const Value &input) {
	auto &config = ClientConfig::GetConfig(context);
	auto parameter = input.ToString();
	config.profiler_save_location = parameter;
}

void ProfileOutputSetting::ResetLocal(ClientContext &context) {
	ClientConfig::GetConfig(context).profiler_save_location = ClientConfig().profiler_save_location;
}

Value ProfileOutputSetting::GetSetting(const ClientContext &context) {
	auto &config = ClientConfig::GetConfig(context);
	return Value(config.profiler_save_location);
}

//===----------------------------------------------------------------------===//
// Profiling Mode
//===----------------------------------------------------------------------===//
void ProfilingModeSetting::SetLocal(ClientContext &context, const Value &input) {
	auto parameter = StringUtil::Lower(input.ToString());
	auto &config = ClientConfig::GetConfig(context);
	if (parameter == "standard") {
		config.enable_profiler = true;
		config.enable_detailed_profiling = false;
	} else if (parameter == "detailed") {
		config.enable_profiler = true;
		config.enable_detailed_profiling = true;

		// add optimizer settings to the profiler settings
		auto optimizer_settings = MetricsUtils::GetOptimizerMetrics();
		for (auto &setting : optimizer_settings) {
			config.profiler_settings.insert(setting);
		}

		// add the phase timing settings to the profiler settings
		auto phase_timing_settings = MetricsUtils::GetPhaseTimingMetrics();
		for (auto &setting : phase_timing_settings) {
			config.profiler_settings.insert(setting);
		}
	} else if (parameter == "all") {
		config.enable_profiler = true;
		auto all_metrics = MetricsUtils::GetAllMetrics();
		for (auto &metric : all_metrics) {
			config.profiler_settings.insert(metric);
		}
	} else {
		throw ParserException("Unrecognized profiling mode \"%s\", supported formats: [standard, detailed]", parameter);
	}
}

void ProfilingModeSetting::ResetLocal(ClientContext &context) {
	ClientConfig::GetConfig(context).enable_profiler = ClientConfig().enable_profiler;
	ClientConfig::GetConfig(context).enable_detailed_profiling = ClientConfig().enable_detailed_profiling;
	ClientConfig::GetConfig(context).emit_profiler_output = ClientConfig().emit_profiler_output;
	ClientConfig::GetConfig(context).profiler_settings = ClientConfig().profiler_settings;
}

Value ProfilingModeSetting::GetSetting(const ClientContext &context) {
	auto &config = ClientConfig::GetConfig(context);
	if (!config.enable_profiler) {
		return Value();
	}
	return Value(config.enable_detailed_profiling ? "detailed" : "standard");
}

//===----------------------------------------------------------------------===//
// Profiling Coverage Setting
//===----------------------------------------------------------------------===//
void ProfilingCoverageSetting::SetLocal(ClientContext &context, const Value &input) {
	auto setting_type = EnumUtil::FromString<ProfilingCoverage>(input.ToString());
	auto &config = ClientConfig::GetConfig(context);
	config.profiling_coverage = setting_type;
}

void ProfilingCoverageSetting::ResetLocal(ClientContext &context) {
	auto &config = ClientConfig::GetConfig(context);
	config.profiling_coverage = ProfilingCoverage::SELECT;
}

Value ProfilingCoverageSetting::GetSetting(const ClientContext &context) {
	const auto &config = ClientConfig::GetConfig(context);
	return Value(EnumUtil::ToString(config.profiling_coverage));
}

//===----------------------------------------------------------------------===//
// Progress Bar Time
//===----------------------------------------------------------------------===//
void ProgressBarTimeSetting::SetLocal(ClientContext &context, const Value &input) {
	auto &config = ClientConfig::GetConfig(context);
	ProgressBar::SystemOverrideCheck(config);
	config.wait_time = input.GetValue<int32_t>();
	config.enable_progress_bar = true;
}

void ProgressBarTimeSetting::ResetLocal(ClientContext &context) {
	auto &config = ClientConfig::GetConfig(context);
	ProgressBar::SystemOverrideCheck(config);
	config.wait_time = ClientConfig().wait_time;
	config.enable_progress_bar = ClientConfig().enable_progress_bar;
}

Value ProgressBarTimeSetting::GetSetting(const ClientContext &context) {
	return Value::BIGINT(ClientConfig::GetConfig(context).wait_time);
}

//===----------------------------------------------------------------------===//
// Schema
//===----------------------------------------------------------------------===//
void SchemaSetting::SetLocal(ClientContext &context, const Value &input) {
	auto parameter = input.ToString();
	auto &client_data = ClientData::Get(context);
	client_data.catalog_search_path->Set(CatalogSearchEntry::Parse(parameter), CatalogSetPathType::SET_SCHEMA);
}

void SchemaSetting::ResetLocal(ClientContext &context) {
	// FIXME: catalog_search_path is controlled by both SchemaSetting and SearchPathSetting
	auto &client_data = ClientData::Get(context);
	client_data.catalog_search_path->Reset();
}

Value SchemaSetting::GetSetting(const ClientContext &context) {
	auto &client_data = ClientData::Get(context);
	return client_data.catalog_search_path->GetDefault().schema;
}

//===----------------------------------------------------------------------===//
// Search Path
//===----------------------------------------------------------------------===//
void SearchPathSetting::SetLocal(ClientContext &context, const Value &input) {
	auto parameter = input.ToString();
	auto &client_data = ClientData::Get(context);
	client_data.catalog_search_path->Set(CatalogSearchEntry::ParseList(parameter), CatalogSetPathType::SET_SCHEMAS);
}

void SearchPathSetting::ResetLocal(ClientContext &context) {
	// FIXME: catalog_search_path is controlled by both SchemaSetting and SearchPathSetting
	auto &client_data = ClientData::Get(context);
	client_data.catalog_search_path->Reset();
}

Value SearchPathSetting::GetSetting(const ClientContext &context) {
	auto &client_data = ClientData::Get(context);
	auto &set_paths = client_data.catalog_search_path->GetSetPaths();
	return Value(CatalogSearchEntry::ListToString(set_paths));
}

//===----------------------------------------------------------------------===//
// Secret Directory
//===----------------------------------------------------------------------===//
void SecretDirectorySetting::SetGlobal(DatabaseInstance *db, DBConfig &config, const Value &input) {
	config.secret_manager->SetPersistentSecretPath(input.ToString());
}

void SecretDirectorySetting::ResetGlobal(DatabaseInstance *db, DBConfig &config) {
	config.secret_manager->ResetPersistentSecretPath();
}

Value SecretDirectorySetting::GetSetting(const ClientContext &context) {
	auto &config = DBConfig::GetConfig(context);
	return config.secret_manager->PersistentSecretPath();
}

//===----------------------------------------------------------------------===//
// Storage Compatibility Version
//===----------------------------------------------------------------------===//
void StorageCompatibilityVersionSetting::SetGlobal(DatabaseInstance *db, DBConfig &config, const Value &input) {
	auto version_string = input.GetValue<string>();
	auto serialization_compatibility = SerializationCompatibility::FromString(version_string);
	config.options.serialization_compatibility = serialization_compatibility;
}

void StorageCompatibilityVersionSetting::ResetGlobal(DatabaseInstance *db, DBConfig &config) {
	config.options.serialization_compatibility = DBConfigOptions().serialization_compatibility;
}

Value StorageCompatibilityVersionSetting::GetSetting(const ClientContext &context) {
	auto &config = DBConfig::GetConfig(context);

	auto &version_name = config.options.serialization_compatibility.duckdb_version;
	return Value(version_name);
}

//===----------------------------------------------------------------------===//
// Streaming Buffer Size
//===----------------------------------------------------------------------===//
void StreamingBufferSizeSetting::SetLocal(ClientContext &context, const Value &input) {
	auto &config = ClientConfig::GetConfig(context);
	config.streaming_buffer_size = DBConfig::ParseMemoryLimit(input.ToString());
}

void StreamingBufferSizeSetting::ResetLocal(ClientContext &context) {
	auto &config = ClientConfig::GetConfig(context);
	config.SetDefaultStreamingBufferSize();
}

Value StreamingBufferSizeSetting::GetSetting(const ClientContext &context) {
	auto &config = ClientConfig::GetConfig(context);
	return Value(StringUtil::BytesToHumanReadableString(config.streaming_buffer_size));
}

//===----------------------------------------------------------------------===//
// Temp Directory
//===----------------------------------------------------------------------===//
void TempDirectorySetting::SetGlobal(DatabaseInstance *db, DBConfig &config, const Value &input) {
	if (!config.options.enable_external_access) {
		throw PermissionException("Modifying the temp_directory has been disabled by configuration");
	}
	config.options.temporary_directory = input.IsNull() ? "" : input.ToString();
	config.options.use_temporary_directory = !config.options.temporary_directory.empty();
	if (db) {
		auto &buffer_manager = BufferManager::GetBufferManager(*db);
		buffer_manager.SetTemporaryDirectory(config.options.temporary_directory);
	}
}

void TempDirectorySetting::ResetGlobal(DatabaseInstance *db, DBConfig &config) {
	if (!config.options.enable_external_access) {
		throw PermissionException("Modifying the temp_directory has been disabled by configuration");
	}
	config.SetDefaultTempDirectory();
	config.options.use_temporary_directory = DBConfigOptions().use_temporary_directory;
	if (db) {
		auto &buffer_manager = BufferManager::GetBufferManager(*db);
		buffer_manager.SetTemporaryDirectory(config.options.temporary_directory);
	}
}

Value TempDirectorySetting::GetSetting(const ClientContext &context) {
	auto &buffer_manager = BufferManager::GetBufferManager(context);
	return Value(buffer_manager.GetTemporaryDirectory());
}

//===----------------------------------------------------------------------===//
// Temporary File Encryption
//===----------------------------------------------------------------------===//
void TempFileEncryptionSetting::SetGlobal(DatabaseInstance *db, DBConfig &config, const Value &input) {
	auto setting = input.GetValue<bool>();
	if (config.options.temp_file_encryption == setting) {
		// setting is the current setting
		return;
	}

	if (db) {
		auto &buffer_manager = BufferManager::GetBufferManager(*db);
		if (buffer_manager.HasFilesInTemporaryDirectory()) {
			throw PermissionException("Existing temporary files found: Modifying the temp_file_encryption setting "
			                          "while there are existing temporary files is disabled.");
		}
	}

	config.options.temp_file_encryption = setting;
}

void TempFileEncryptionSetting::ResetGlobal(DatabaseInstance *db, DBConfig &config) {
	if (config.options.temp_file_encryption == true) {
		// setting is the current setting
		return;
	}

	if (db) {
		auto &buffer_manager = BufferManager::GetBufferManager(*db);
		if (buffer_manager.HasFilesInTemporaryDirectory()) {
			throw PermissionException("Existing temporary files found: Modifying the temp_file_encryption setting "
			                          "while there are existing temporary files is disabled.");
		}
	}

	config.options.temp_file_encryption = true;
}

Value TempFileEncryptionSetting::GetSetting(const ClientContext &context) {
	auto &config = DBConfig::GetConfig(context);
	return Value::BOOLEAN(config.options.temp_file_encryption);
}

//===----------------------------------------------------------------------===//
// Threads
//===----------------------------------------------------------------------===//
void ThreadsSetting::SetGlobal(DatabaseInstance *db, DBConfig &config, const Value &input) {
	auto new_val = input.GetValue<int64_t>();
	if (new_val < 1) {
		throw SyntaxException("Must have at least 1 thread!");
	}
	auto new_maximum_threads = NumericCast<idx_t>(new_val);
	if (db) {
		TaskScheduler::GetScheduler(*db).SetThreads(new_maximum_threads, config.options.external_threads);
	}
	config.options.maximum_threads = new_maximum_threads;
}

void ThreadsSetting::ResetGlobal(DatabaseInstance *db, DBConfig &config) {
	idx_t new_maximum_threads = config.GetSystemMaxThreads(*config.file_system);
	if (db) {
		TaskScheduler::GetScheduler(*db).SetThreads(new_maximum_threads, config.options.external_threads);
	}
	config.options.maximum_threads = new_maximum_threads;
}

Value ThreadsSetting::GetSetting(const ClientContext &context) {
	auto &config = DBConfig::GetConfig(context);
	return Value::BIGINT(NumericCast<int64_t>(config.options.maximum_threads));
}

//===----------------------------------------------------------------------===//
// Username
//===----------------------------------------------------------------------===//
void UsernameSetting::SetGlobal(DatabaseInstance *db, DBConfig &config, const Value &input) {
	// nop
}

void UsernameSetting::ResetGlobal(DatabaseInstance *db, DBConfig &config) {
	// nop
}

Value UsernameSetting::GetSetting(const ClientContext &context) {
	return Value();
}

} // namespace duckdb
